package com.cloud.message;

import com.cloud.exception.CustomException;
import com.cloud.utils.JSONUtils;
import org.apache.commons.lang3.StringUtils;
import org.java_websocket.client.WebSocketClient;
import org.java_websocket.drafts.Draft_6455;
import org.java_websocket.handshake.ServerHandshake;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.net.URI;

/**
 * WebSocketClient配置
 *
 * @author yzj
 */
public abstract class AbstractSocketConsumer<T> {

    private static Logger logger = LoggerFactory.getLogger(AbstractSocketConsumer.class);

    public AbstractSocketConsumer() {
    }

    /**
     * 初始化WebSocket客户端
     *
     * @param clientId
     * @return
     */
    public void InItSocketClient(String serverHost, String clientId) {
        try {
            if (WebSocketConsumer.checkClientExist(clientId)) {
                String host = serverHost + "/" + clientId;
                WebSocketClient socketClient = new WebSocketClient(new URI(host), new Draft_6455()) {
                    @Override
                    public void onOpen(ServerHandshake handshake) {
                        logger.info("[SocketClient消息]连接成功");
                    }

                    @Override
                    public void onMessage(String message) {
                        logger.info("[SocketClient消息]接收消息:{}", message);
                        getMessage(parseMessage(message), message);
                    }

                    @Override
                    public void onClose(int code, String reason, boolean remote) {
                        logger.info("[SocketClient消息]退出连接");
                        try {
                            // 移除WebSocketClient对象
                            WebSocketConsumer.removeSocketClient(clientId);
                            Thread.sleep(30000);
                            InItSocketClient(serverHost, clientId);
                            logger.info("[SocketClient消息]重新连接中...");
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                    }

                    @Override
                    public void onError(Exception ex) {
                        logger.info("[SocketClient消息]连接错误:{}", ex.getMessage());
                    }
                };
                socketClient.connect();
                WebSocketConsumer.setSocketClient(clientId, socketClient);
            } else {
                logger.info("SocketClient对象已创建,ClientId:" + clientId);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 获取消息内容
     *
     * @param message
     */
    public abstract String getMessage(T object, String message);

    /**
     * 反序列化解析消息
     *
     * @param message 消息体
     * @return 序列化结果
     */
    protected T parseMessage(String message) {
        if (StringUtils.isEmpty(message)) {
            return null;
        }
        final Type type = this.getMessageType();
        if (type instanceof Class) {
            try {
                return JSONUtils.toObject(new String(message.getBytes()), (Class<T>) type);
            } catch (Exception e) {
                logger.error("解析消息对象失败:{}", e.getMessage());
            }
        } else {
            logger.warn("解析消息错误.{}", message);
        }
        return null;
    }

    protected Type getMessageType() {
        Type superType = this.getClass().getGenericSuperclass();
        if (superType instanceof ParameterizedType) {
            ParameterizedType parameterizedType = (ParameterizedType) superType;
            Type[] actualTypeArguments = parameterizedType.getActualTypeArguments();
            if (actualTypeArguments.length == 1) {
                return actualTypeArguments[0];
            } else {
                throw new CustomException("类型参数的数量必须为1");
            }
        } else {
            // 如果没有定义泛型,解析为Object
            return Object.class;
        }
    }

}
